package cc.verywell.pureblock.msg.nfs.idx.extraDownload;

import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.security.DigestException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;

import cc.verywell.pureblock.func.node.KeyShell;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.net.AioProvider;
import cc.verywell.pureblock.util.sign.MD5;

/**
* 
* Copyright © 2020 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2020年2月3日 下午2:15:56
*/
public class DloTargetCheck {

	public static int DEFAULTREAD = AioProvider.FILE_MAX_SIZE/10;//默认50k用于检查MD5,请根据实际情况最大加到512k(AioProvider.FILE_MAX_SIZE);
	public static final String HASH_NAME = MD5.HASH_NAME;
	public static final ThreadLocal<MD5> hash = new ThreadLocal<MD5>() {
		@Override
		protected MD5 initialValue() {
			return new MD5();
		}
	};
	
	public final FileChannel channel;
	public final DloMap map;
	public final boolean isEmptyPath;//true:这是一个空转类
	public DloTargetCheck(boolean isEmptyPath,FileChannel channel,DloMap map) {
		this.isEmptyPath = isEmptyPath;
		this.channel = channel;
		this.map = map;
	}
	final private AtomicBoolean stoped = new AtomicBoolean(false);
	final private Semaphore lock = new Semaphore(1); ///用来lock writeBuffer不会给重复调用;
	final private BlockingQueue<TargetCheck> stack = new LinkedBlockingQueue<>();
	static public class TargetCheck{
		public final KeyShell key;
		public final byte[] hash;
		public final DltoolPosLen pos;
		public final boolean isDataStock;
		public TargetCheck(KeyShell key, byte[] hash, DltoolPosLen pos, boolean isDataStock) {
			this.key = key;
			this.hash = hash;
			this.pos = pos;
			this.isDataStock = isDataStock;
		}
	}
	
	//对外接口组
	public void join(TargetCheck newcheck) throws InterruptedException,IOException, DigestException {
		if(stoped.get())throw new InterruptedException("target check closed.");
		stack.put(newcheck);
		/*
		//当A运行到这里时
		if(lock.tryAcquire()) {
			try {
				TargetCheck check = null;
				while(null!=(check=stack.poll())) {
					check(check);
					if(stoped.get())return;
				}
				//B运行到这里，侧stack会无人调用;
			}finally {
				lock.release();
			}
		}
		*/
		do{
			if(lock.tryAcquire()) {
				try {
					if(stoped.get())return;
					TargetCheck check = stack.poll();
					if(null!=check)
						check(check);
				}finally {
					lock.release();
				}
			}else {
				return;
			}
		}while(null!=stack.peek());
		this.map.checkFinish(null);
	}
	private ByteBuffer bb;
	private MessageDigest messageDigest;
	private byte[] hashout;
	private void check(TargetCheck check) throws IOException, DigestException {
		if(isEmptyPath) {//true:这是一个空转类
			map.add_datastock_by_errcheck(check.pos.len);
			return;
		}
		if(bb==null) {
			try {
				messageDigest = MessageDigest.getInstance(HASH_NAME);
			} catch (NoSuchAlgorithmException e) {
				if(Pblog.INIT)Pblog.INIT(e);
				map.add_readonly_req_by_errcheck(check.key,check.pos);
				throw new RuntimeException(e);
			}
			hashout = new byte[messageDigest.getDigestLength()];
			bb = ByteBuffer.allocate(DEFAULTREAD);
		}else {
			bb.clear();
		}
		long pos = check.pos.pos,len = check.pos.len;
		try(
				FileLock filelock = channel.lock(pos,len,true);
		){
			bb.position(bb.limit());
			while(len>0) {
				bb.compact();
				if(bb.remaining()>len)bb.limit((int)(bb.limit()-bb.remaining()+len));
				int read = channel.read(bb, pos);
				if(read<0)throw new EOFException();
				pos+=read;len-=read;
				bb.flip();
				messageDigest.update(bb);
			}
			while(bb.hasRemaining())
				messageDigest.update(bb);
			if(messageDigest.digest(hashout, 0, hashout.length)!=hashout.length) {
				throw new RuntimeException("md5 length not is "+hashout.length);
			}
		}finally {
			messageDigest.reset();
		}
			if(MessageDigest.isEqual(hashout, check.hash)) {
				map.add_datastock_by_errcheck(check);
			}else {
				map.add_readonly_req_by_errcheck(check.key,check.pos);
			}
	}
	
	
	//流程控制
	public void stop() {
		if(stoped.compareAndSet(false, true)) {
//			启用会引起死锁
//			try {
//				lock.acquire();
//				lock.release();
//			} catch (InterruptedException e) {}
		}
	};
// 下面如果启用会引起死锁
	public void waitFinish() {
		try {
			lock.acquire();
			lock.release();
		} catch (InterruptedException e) {}
	};
//	public void waitFinishAndStop() {
//		try {
//			lock.acquire();
//			stoped.compareAndSet(false, true);
//			lock.release();
//		} catch (InterruptedException e) {}
//	};
}
